package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: xyl
 * @Description: TODO
 * @Date: 2023/11/22 14:30
 */
public class ZjsTStockPledge {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("zjs_t_stock_pledge").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_stock_pledge");
        long zs_count = zs_data.count();
        if(zs_count>0){
            //修改 --标的企业
            spark.sql("insert overwrite table a_dc_ep_incr.zjs_t_stock_pledge select " +
                    "md5(concat_ws('',MD5(CONCAT('ExportFromMongo_',qcc1.company_id)),translate(translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\"),' ',''),translate(translate(translate(zs.stk_pawn_czper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(zs.stk_pawn_zqper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(qcc1.company_name,\"（\",\"(\"),\"）\",\")\"),' ',''))) as id," +
                    "MD5(CONCAT('ExportFromMongo_',qcc1.company_id))," +
                    "qcc1.company_name," +
                    "translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\")," +
                    "MD5(CONCAT('ExportFromMongo_',qcc2.company_id))," +
                    "zs.stk_pawn_czper," +
                    "zs.stk_pawn_czcerno," +
                    "zs.stk_pawn_czamt," +
                    "MD5(CONCAT('ExportFromMongo_',qcc3.company_id))," +
                    "zs.stk_pawn_zqper," +
                    "zs.stk_pawn_zqcerno," +
                    "MD5(CONCAT('ExportFromMongo_',qcc1.company_id))," +
                    "qcc1.company_name," +
                    "qcc1.credit_code," +
                    "zs.stk_pawn_regdate," +
                    "zs.stk_pawn_status," +
                    "zs.stk_pawn_date," +
                    "zsalt.stk_pawn_bgnr," +
                    "zsalt.stk_pawn_bgrq," +
                    "zsrev.stk_pawn_date," +
                    "zsrev.stk_pawn_res," +
                    "zs.jobid as dates," +
                    "'0' as isadd  \n" +
                    "from a_dc_ep_ods.zs_t_stock_pledge zs \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_stock_pledge_del where entid is not null and entid!='')zsdel on zs.entid=zsdel.entid \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc1 on upper(trim(zsdel.creditcode)) = upper(trim(qcc1.credit_code))  \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_alt where entid is not null and entid!='')zsalt on zs.url=zsalt.url \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_rev where entid is not null and entid!='')zsrev on zs.url=zsrev.url  \n" +
                    "left join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc2 on upper(trim(zs.stk_pawn_czcerno)) = upper(trim(qcc2.credit_code))  \n" +
                    "left join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc3 on upper(trim(zs.stk_pawn_zqcerno)) = upper(trim(qcc3.credit_code))");

            //修改 --出质企业
            spark.sql("insert into table a_dc_ep_incr.zjs_t_stock_pledge select " +
                    "md5(concat_ws('',MD5(CONCAT('ExportFromMongo_',qcc2.company_id)),translate(translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\"),' ',''),translate(translate(translate(zs.stk_pawn_czper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(zs.stk_pawn_zqper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(qcc1.company_name,\"（\",\"(\"),\"）\",\")\"),' ',''))) as id," +
                    "MD5(CONCAT('ExportFromMongo_',qcc2.company_id))," +
                    "qcc2.company_name," +
                    "translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\")," +
                    "MD5(CONCAT('ExportFromMongo_',qcc2.company_id))," +
                    "zs.stk_pawn_czper," +
                    "zs.stk_pawn_czcerno," +
                    "zs.stk_pawn_czamt," +
                    "MD5(CONCAT('ExportFromMongo_',qcc3.company_id))," +
                    "zs.stk_pawn_zqper," +
                    "zs.stk_pawn_zqcerno," +
                    "MD5(CONCAT('ExportFromMongo_',qcc1.company_id))," +
                    "qcc1.company_name," +
                    "qcc1.credit_code," +
                    "zs.stk_pawn_regdate," +
                    "zs.stk_pawn_status," +
                    "zs.stk_pawn_date," +
                    "zsalt.stk_pawn_bgnr," +
                    "zsalt.stk_pawn_bgrq," +
                    "zsrev.stk_pawn_date," +
                    "zsrev.stk_pawn_res," +
                    "zs.jobid as dates," +
                    "'0' as isadd  \n" +
                    "from a_dc_ep_ods.zs_t_stock_pledge zs \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_stock_pledge_del where entid is not null and entid!='')zsdel on zs.entid=zsdel.entid \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc1 on upper(trim(zsdel.creditcode)) = upper(trim(qcc1.credit_code))  \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_alt where entid is not null and entid!='')zsalt on zs.url=zsalt.url \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_rev where entid is not null and entid!='')zsrev on zs.url=zsrev.url  \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc2 on upper(trim(zs.stk_pawn_czcerno)) = upper(trim(qcc2.credit_code))  \n" +
                    "left join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc3 on upper(trim(zs.stk_pawn_zqcerno)) = upper(trim(qcc3.credit_code))");

            //修改 --质权企业
            spark.sql("insert into table a_dc_ep_incr.zjs_t_stock_pledge select " +
                    "md5(concat_ws('',MD5(CONCAT('ExportFromMongo_',qcc3.company_id)),translate(translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\"),' ',''),translate(translate(translate(zs.stk_pawn_czper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(zs.stk_pawn_zqper,\"（\",\"(\"),\"）\",\")\"),' ',''),translate(translate(translate(qcc1.company_name,\"（\",\"(\"),\"）\",\")\"),' ',''))) as id," +
                    "MD5(CONCAT('ExportFromMongo_',qcc3.company_id))," +
                    "qcc3.company_name," +
                    "translate(translate(translate(translate(zs.stk_pawn_regno,\"（\",\"(\"),\"）\",\")\"),\"【\",\"[\"),\"】\",\"]\")," +
                    "MD5(CONCAT('ExportFromMongo_',qcc2.company_id))," +
                    "zs.stk_pawn_czper," +
                    "zs.stk_pawn_czcerno," +
                    "zs.stk_pawn_czamt," +
                    "MD5(CONCAT('ExportFromMongo_',qcc3.company_id))," +
                    "zs.stk_pawn_zqper," +
                    "zs.stk_pawn_zqcerno," +
                    "MD5(CONCAT('ExportFromMongo_',qcc1.company_id))," +
                    "qcc1.company_name," +
                    "qcc1.credit_code," +
                    "zs.stk_pawn_regdate," +
                    "zs.stk_pawn_status," +
                    "zs.stk_pawn_date," +
                    "zsalt.stk_pawn_bgnr," +
                    "zsalt.stk_pawn_bgrq," +
                    "zsrev.stk_pawn_date," +
                    "zsrev.stk_pawn_res," +
                    "zs.jobid as dates," +
                    "'0' as isadd  \n" +
                    "from a_dc_ep_ods.zs_t_stock_pledge zs \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_stock_pledge_del where entid is not null and entid!='')zsdel on zs.entid=zsdel.entid \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc1 on upper(trim(zsdel.creditcode)) = upper(trim(qcc1.credit_code))  \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_alt where entid is not null and entid!='')zsalt on zs.url=zsalt.url \n" +
                    "left join (select * from a_dc_ep_ods.zs_t_stock_pledge_rev where entid is not null and entid!='')zsrev on zs.url=zsrev.url  \n" +
                    "left join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc2 on upper(trim(zs.stk_pawn_czcerno)) = upper(trim(qcc2.credit_code))  \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code!='') qcc3 on upper(trim(zs.stk_pawn_zqcerno)) = upper(trim(qcc3.credit_code))");

            //删除
            spark.sql("insert into table a_dc_ep_incr.zjs_t_stock_pledge " +
                    "select t1.id,t1.company_id,t1.company_name,t1.stk_pawn_regno,t1.stk_pawn_czcompany_id,t1.stk_pawn_czper,t1.stk_pawn_czcerno,t1.stk_pawn_czamt,t1.stk_pawn_zqcompany_id,t1.stk_pawn_zqper,t1.stk_pawn_zqcerno,t1.stk_pawn_bdcompany_id,t1.stk_pawn_bdper,t1.stk_pawn_bdcerno,t1.stk_pawn_regdate," +
                    "t1.stk_pawn_status,t1.publicity_date,t1.stk_pawn_bgnr,t1.stk_pawn_bgrq,t1.write_off_date,t1.stk_pawn_res,t3.jobid,'-1'  \n" +
                    "from a_dc_ep_dwi.zjs_t_stock_pledge t1 \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where company_id is not null and company_id!='')t2 on t1.company_id=MD5(CONCAT('ExportFromMongo_',t2.company_id)) \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_stock_pledge_del where creditcode is not null and creditcode!='' )t3 on upper(trim(t3.creditcode)) = upper(trim(t2.credit_code)) ");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.zjs_t_stock_pledge");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
